# pip install fastapi[all]
# pip install imageio-ffmpeg
# pip install ffmpeg-python
# pip install requests
# pip install scenedetect
# pip install opencv-python-headless

import os
import ffmpeg
import asyncio
import logging
import requests
import hashlib
from datetime import datetime
from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel
from scenedetect import detect, ContentDetector
from typing import List, Dict, Optional

from tiktok_downloader import VideoDownloader
from filter_video import FilterVideo
from google_speech import GoogleSpeechTranscriber

# ============================
# 配置日志
# ============================
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

# ============================
# FastAPI 初始化
# ============================
app = FastAPI()
SECRET_KEY = "ghijklmnopq"

# 任务锁（防止并发执行）
task_lock = asyncio.Lock()

# 本地目录
OUTPUT_MP3_DIR = "/home/data/mp3"
OUTPUT_IMG_DIR = "/home/data/image"
DOWNLOAD_DIR = "/home/data/mp4"
os.makedirs(OUTPUT_MP3_DIR, exist_ok=True)
os.makedirs(OUTPUT_IMG_DIR, exist_ok=True)
os.makedirs(DOWNLOAD_DIR, exist_ok=True)


class AIRequest(BaseModel):
    timestamp: str
    rm: str
    key: str
    text: str
    type: Optional[str] = None
    prompt: Optional[str] = None
    from_text: Optional[str] = None
    history: Optional[List[Dict[str, str]]] = None


# ============================
# 工具函数
# ============================
def is_url(path: str) -> bool:
    return path.startswith("http://") or path.startswith("https://")


def get_audio_duration(audio_path: str) -> float:
    probe = ffmpeg.probe(audio_path)
    return float(probe['format']['duration'])


def call_java_api(status: str, data: dict):
    java_api_url = "http://serverapp:8080/video-analysis/public/fghijklmnop/change-status"
    try:
        requests.post(java_api_url, json={"status": status, "data": data}, timeout=5)
    except Exception as e:
        logging.error(f"调用 Java API 失败: {e}")


# ============================
# 核心流程
# ============================
async def download_file(video_url: str, video_id: str) -> Optional[str]:
    """异步下载视频"""
    video_filename = FilterVideo.get_video_filename(video_url)
    call_java_api("DOWNLOADING", {"video_id": video_id})

    downloader = VideoDownloader(save_path=DOWNLOAD_DIR)
    video_path = await downloader.download_video(video_url, custom_name=video_filename)

    if video_path and len(video_path) >= 10:
        video_time = get_audio_duration(video_path)
        call_java_api("DOWNLOAD", {"video_id": video_id, "video_path": video_path, "video_time": video_time})
        return video_path

    call_java_api("ERROR", {"video_id": video_id, "errormsg": "视频下载失败"})
    return None


def extract_audio_and_frames(video_path: str, video_id: str):
    """提取音频和关键帧"""
    audio_path = os.path.join(OUTPUT_MP3_DIR, f"{video_id}.mp3")
    ffmpeg.input(video_path).output(audio_path, format="mp3", acodec="libmp3lame").overwrite_output().run()

    keyframes = []  # 关键帧提取逻辑可按需启用
    call_java_api("EXTRACT", {"video_id": video_id, "mp3": audio_path, "frames": keyframes})
    return audio_path, keyframes


def ai_pipeline(audio_path: str, video_id: str, language: str):
    """调用 AI 解析"""
    try:
        transcriber = GoogleSpeechTranscriber("zeta-courage-452114-u1")
        result = transcriber.transcribe_audio(audio_path, language)
        call_java_api("END", {"video_id": video_id, "ai_data": result})
    except Exception as e:
        logging.error(f"AI 解析失败: {e}")
        call_java_api("ERROR", {"video_id": video_id, "errormsg": str(e)})


async def video_pipeline(url: str, video_id: str, language: str):
    """完整视频处理流程"""
    async with task_lock:
        try:
            if is_url(url):
                video_path = await download_file(url, video_id)
            else:
                video_path = os.path.join(DOWNLOAD_DIR, url.lstrip("/"))
                video_time = get_audio_duration(video_path)
                call_java_api("DOWNLOAD", {"video_id": video_id, "video_time": video_time})

            if not video_path:
                return

            audio_path, _ = extract_audio_and_frames(video_path, video_id)
            if not audio_path:
                return

            ai_pipeline(audio_path, video_id, language)
        finally:
            logging.info("任务结束")


# ============================
# API 路由
# ============================
@app.get("/process-video")
async def process_video(url: str, video_id: str, language: str, background_tasks: BackgroundTasks):
    if task_lock.locked():
        return {"status": False, "message": "任务正在进行"}

    background_tasks.add_task(video_pipeline, url, video_id, language)
    return {"status": True, "message": "任务已提交"}


@app.get("/process-local-video")
async def process_local_video(video_path: str, video_id: str, language: str, background_tasks: BackgroundTasks):
    if task_lock.locked():
        return {"status": False, "message": "任务正在进行"}

    background_tasks.add_task(video_pipeline, video_path, video_id, language)
    return {"status": True, "message": "任务已提交"}


@app.post("/ai-process")
async def ai_process(request: AIRequest):
    # 签名校验
    data_list = [str(request.timestamp), request.rm, SECRET_KEY]
    data_list.sort()
    computed_signature = hashlib.sha256("".join(data_list).encode()).hexdigest()

    if computed_signature != request.key:
        raise HTTPException(status_code=404, detail="Signature verification failed")

    transcriber = GoogleSpeechTranscriber("zeta-courage-452114-u1")
    result = transcriber.transcribe(request.text, request.prompt, request.type, request.history)
    return {"status": 200, "message": result}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000, reload=True)
